package mutiyi;

import org.eclipse.paho.client.mqttv3.*;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.MessageChannels;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;

/**
 * Author:yang
 * Date:2024-03-06 14:59
 */
@Configuration
@EnableIntegration
public class BServer {

    private static final String INBOUND_TOPIC = "testHeartbeat";
    private static final String OUTBOUND_TOPIC = "testHeartbeat_reply";
    @Bean
    public DirectChannel mqttInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public IntegrationFlow mqttInFlow() {
        return IntegrationFlows.from(mqttInbound())
                .channel(mqttInputChannel())
                .handle(handler())
                .get();
    }

    @Bean
    public MqttPahoMessageDrivenChannelAdapter mqttInbound() {
        String broker = "tcp://81.71.46.14:1882";
        String clientId = "B";
        String topic = "testHeartbeat";

        DefaultMqttPahoClientFactory clientFactory = new DefaultMqttPahoClientFactory();
        clientFactory.setConnectionOptions(createMqttConnectOptions());

        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(broker, clientId, INBOUND_TOPIC);
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return new MessageHandler() {
            @Override
            public void handleMessage(Message<?> message) {
                String payload = message.getPayload().toString();
                System.out.println("A的消息: " + payload);

                // 处理完消息后回信
                sendMessageToA("B的消息: HelloA");
            }
        };
    }



    private void sendMessageToA(String message) {
        // 实现发送消息
        try {
            String aBroker = "tcp://81.71.46.14:1882";
            String aClientId = "BToA";
            String aTopic = "testHeartbeat";

            MqttClient mqttClient = new MqttClient(aBroker, aClientId);
            MqttConnectOptions connectOptions = new MqttConnectOptions();
            connectOptions.setCleanSession(true);
            connectOptions.setUserName("test1");
            connectOptions.setPassword("test1".toCharArray());

            mqttClient.connect(connectOptions);

            MqttMessage mqttMessage = new MqttMessage(message.getBytes());
            mqttClient.publish(aTopic, mqttMessage);
            mqttClient.disconnect();
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    private MqttConnectOptions createMqttConnectOptions() {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName("test1");
        options.setPassword("test1".toCharArray());
        return options;
    }

    public static void main(String[] args) {
        AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(BServer.class);
        context.registerShutdownHook();
    }
}
